package com.chukun.tx.message.task;

import com.chukun.tx.constants.MessageConstant;
import com.chukun.tx.mapper.MqMessageMapper;
import com.chukun.tx.message.producer.MsgProducer;
import com.chukun.tx.message.producer.OrderProducer;
import com.chukun.tx.message.queue.DefaultMessageProcessor;
import com.chukun.tx.message.queue.DefaultTimeWheelProcessor;
import com.chukun.tx.message.queue.MessageProcessor;
import com.chukun.tx.model.MqMessage;
//import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.*;

@Component
public class MessageSendTask {

    @Autowired
    private MsgProducer msgProducer;
    @Autowired
    private OrderProducer orderProducer;
    @Autowired
    private MqMessageMapper messageMapper;

    private ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1, r -> {
        Thread t = new Thread(r);
        t.setName("messageSend thread");
        t.setDaemon(true);
        return t;
    }, new ThreadPoolExecutor.CallerRunsPolicy());

    @PostConstruct
    public void sendMessage() {
        executorService.scheduleAtFixedRate(() -> {
            MessageProcessor defaultProcessor = DefaultMessageProcessor.getDefaultProcessor();
            MqMessage msg = defaultProcessor.getMsg();
            // 首先将消息置为处理中
            MqMessage updateMsg = new MqMessage();
            BeanUtils.copyProperties(msg, updateMsg);
            updateMsg.setStatus(MessageConstant.MessageStatus.MESSAGE_HANDLING.getCode());
            messageMapper.updateMessage(updateMsg);
//            Message message = new Message();
//            message.setTopic(msg.getTopic());
//            message.setTags(msg.getTag());
//            message.setBody(msg.getContent().getBytes());
//            boolean isSend = orderProducer.sendMessage(message);
            msgProducer.send2TopicTestAQueue(msg.getContent());

            updateMsg.setStatus(MessageConstant.MessageStatus.MESSAGE_CONSUMED.getCode());
            messageMapper.updateMessage(updateMsg);

//            try {
//                int times = msg.getHaveDeadTimes()
//                        * MessageConstant.TIME_INTERVAL[msg.getHaveDeadTimes()];
//                msg.setHaveDeadTimes(msg.getHaveDeadTimes() + 1);
//                msg.setNextExpireTime(System.currentTimeMillis() + 1000 * times);
//                DefaultTimeWheelProcessor.getDefaultProcessor().putFailMessage(msg);
//            } catch (Exception e) {
//
//            } finally {
//                updateMsg.setHaveDeadTimes(msg.getHaveDeadTimes() + 1);
//                updateMsg.setStatus(MessageConstant.MessageStatus.MESSAGE_CONSUME_FAIL.getCode());
//                messageMapper.updateMessage(updateMsg);
//            }
        }, 100, 200, TimeUnit.MILLISECONDS);
    }


}
